package main

import (
	"context"
	"encoding/json"
	"fmt"
	"os"

	//"strconv"
	"sync"
	"time"

	log "github.com/sirupsen/logrus"

	sarama "github.com/Shopify/sarama"
)

type consumerGroupHandler struct {
	name string
}

func (consumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
func (consumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (h consumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	data := make(chan *sarama.ConsumerMessage, 5)

	var c Ceph
	c.CephInit()
	i := 0
	var TotalEndTime int64
	var TotalStartTime int64

	p := "./log/test.log"
	logfile, err := os.OpenFile(p, os.O_RDWR|os.O_CREATE, 0666)
	if err != nil {
		fmt.Println(err)
	}
	defer logfile.Close()
	log.SetOutput(logfile)
	for i := 0; i < 5; i++ {
		go CulTask(data, c, i)
	}

	for msg := range claim.Messages() {
		if msg.Value == nil {
			fmt.Println("nil msg")
		} else {
			data <- msg
			sess.MarkMessage(msg, "")
			sess.Commit()
			i++
			if i == 60 {
				TotalEndTime = time.Now().UnixNano()
				fmt.Println("总计用时", float64((TotalEndTime-TotalStartTime)/1e6))
			}
		}
	}
	c.Free()
	return nil
}

func CulTask(mmsg chan *sarama.ConsumerMessage, c Ceph, i int) {

	fmt.Println("初始化线程", i)
	for {
		msg := <-mmsg
		fmt.Println("线程", i, "计算", msg.Offset)
		log.Printf("thread %d start %d \n", i, msg.Offset)
		time.Sleep(1 * time.Second) //
		var q TQuintet
		data := make([]byte, 4096*1024)
		datat := make([]byte, 4096*1024)
		fmt.Println(msg.Offset)
		json.Unmarshal(msg.Value, &q)
		fmt.Println("读取五元组 ", q.Quin[0].BlockId)
		var wg sync.WaitGroup
		startTime := time.Now().UnixNano()
		for k, v := range q.Quin {
			if Cluster == v.Compute { //计算
				wg.Add(1)
				go func(wg *sync.WaitGroup, k int, v Quintet) {
					defer wg.Done()
					if len(v.Trans) == 3 {
						// data1 := c.getBlockData(string(v.Trans[0].Hash))
						// data2 := c.getBlockData(string(v.Trans[1].Hash))
						// data3 := c.getBlockData(string(v.Trans[2].Hash))
						// data1 := c.GetBlockDataFromHttp(v.Trans[0])
						// data2 := c.GetBlockDataFromHttp(v.Trans[1])
						// data3 := c.GetBlockDataFromHttp(v.Trans[2])
						data1, err := c.GetBlockFromTLS(string(v.Trans[0].Hash), int(v.Trans[0].From))
						if err != nil || data1 == nil {
							log.Println("cluster", Cluster, ":", "获取数据失败:", v.Trans[0].Hash)
							return
						}
						data2, err := c.GetBlockFromTLS(string(v.Trans[1].Hash), int(v.Trans[1].From))
						if err != nil || data1 == nil {
							log.Println("cluster", Cluster, ":", "获取数据失败:", v.Trans[1].Hash)
							return
						}
						data3, err := c.GetBlockFromTLS(string(v.Trans[2].Hash), int(v.Trans[2].From))
						if err != nil || data1 == nil {
							log.Println("cluster", Cluster, ":", "获取数据失败:", v.Trans[2].Hash)
							return
						}
						Xor4M(data1, data2, datat)
						Xor4M(datat, data3, data)
						err = WriteToOtherClusters(ClusterIp(int(v.Storage[0])), ClusterPort, "_4"+string(v.Trans[0].Hash)+"+"+string(v.Trans[1].Hash)+"+"+string(v.Trans[2].Hash), data) //4_表示冗余块
						if err != nil {
							fmt.Println("error when write to other clusters")
						}
					}
				}(&wg, k, v)
			} else {
				for _, vv := range v.Storage { //传输
					if Cluster == vv {

					}
				}
			}

		}
		wg.Wait()
		endTime := time.Now().UnixNano()
		fmt.Println(msg.Offset, q.Quin[0].BlockId, "冗余块计算完毕,耗时", float64((endTime-startTime)/1e6))
	}

}

func DownloadMessage() {
	fmt.Println("consumer启动")

	config := sarama.NewConfig()
	config.Consumer.Return.Errors = false
	config.Version = sarama.V0_11_0_2
	config.Consumer.Offsets.AutoCommit.Enable = false
	config.Consumer.Offsets.Initial = sarama.OffsetNewest

	// group, err := sarama.NewConsumerGroup([]string{KafkaIpAddress}, "t", config)
	group, err := sarama.NewConsumerGroup([]string{ClusterIp(Cluster) + KafkaPort}, "t3", config)

	//sarama.NewConsumerGroupFromClient()
	if err != nil {
		panic(err)
	}
	defer group.Close()

	for {
		handler := consumerGroupHandler{name: "sera"}
		err := group.Consume(context.Background(), []string{"TQuintetLine"}, handler)
		if err != nil {
			fmt.Println(err.Error())
		}
	}
}
